package com.allyes.flume.interceptor.parser;

import com.allyes.adx.log.*;
import com.allyes.flume.interceptor.DynamicRawlogEventHeader;
import com.allyes.flume.interceptor.InvalidLogHandler;
import com.allyes.flume.interceptor.rawlogdigger.DiggerInfoFetcher;
import com.allyes.flume.source.Pair;

import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.IOException;


/**
 * Created by gc on 7/14/16.
 */
public class LogParserAdx extends LogParser {
  private static final String PRODUCT_NAME = "adx";
  private static final Logger logger = LoggerFactory.getLogger(LogParserAdx.class);

  public LogParserAdx() {
    invalidLogHandler = null;
    dynamicRawlogEventHeader = null;
  }

  public LogParserAdx(InvalidLogHandler h, DynamicRawlogEventHeader dh) {
    invalidLogHandler = h;
    dynamicRawlogEventHeader = dh;
  }

  @Override
  public Pair<Boolean, Event> parseToEvent(
      long inode,
      String fileName,
      int lineIndex,
      String line) throws IOException {
    Pair<Boolean, Event> ret = new Pair<Boolean, Event>();

    try {
      RawLog rawlog = toRawlog(line, fileName, lineIndex);
      byte[] serializedRawLog = converToBinary(rawlog);
      Event event = EventBuilder.withBody(serializedRawLog);

      if (dynamicRawlogEventHeader != null) {
        dynamicRawlogEventHeader.updateHeader(event, new DiggerInfoFetcher(rawlog));
      }

      ret.left = true;
      ret.right = event;
      return ret;

    } catch (ParserException.InvalidLogException e) {
      ret.left = false;
      if (invalidLogHandler != null) {
        ret.right = invalidLogHandler.handle(inode, fileName, lineIndex, line, e.getMessage());
      } else {
        ret.right = null;
      }
      logger.error(e + " in (file=" + fileName + ", inode=" + inode + ", lineIndex=" + lineIndex + ")");
      return ret;
    }
  }

  public com.allyes.adx.log.RawLog toRawlog(String line, String filename, int lineno)
      throws ParserException.LogFormatException, ParserException.LogTypeException {
    RawLog rawLog = new RawLog();

    String[] ss =  line.split("\t", -1);
    if (ss.length >= 30) {
      int pos = 0;
      rawLog.setTimestamp(getTimestamp(ss[pos++]));
      rawLog.setLogType(getLogType(ss[pos++]));
      rawLog.setProductName(PRODUCT_NAME);
      rawLog.setDb(ss[pos++]);
      rawLog.setTrafficSource(getTrafficSource(ss[pos++]));
      rawLog.setCreativeType(getCreativeType(ss[pos++]));
      rawLog.setDistribution(getDistribution(ss[pos++]));
      rawLog.setDistributionType(getDistributionType(ss[pos++]));
      rawLog.setImpressionId(ss[pos++]);
      rawLog.setExtImpressionId(ss[pos++]);
      rawLog.setAllyesid(ss[pos++]);
      rawLog.setUserAgent(ss[pos++]);
      rawLog.setIp(ss[pos++]);
      rawLog.setRequestUrl(ss[pos++]);
      rawLog.setRefererUrl(ss[pos++]);
      rawLog.setDeviceType(Integer.valueOf(ss[pos++]));
      rawLog.setGeoId(ss[pos++]);
      rawLog.setIfa(ss[pos++]);
      rawLog.setAdspaceId(ss[pos++]);
      rawLog.setCreativeId(ss[pos++]);
      rawLog.setSellerId(Integer.valueOf(ss[pos++]));
      rawLog.setBuyerId(Integer.valueOf(ss[pos++]));
      rawLog.setDealId(Integer.valueOf(ss[pos++]));
      rawLog.setBidFloor(Integer.valueOf(ss[pos++]));
      rawLog.setBidFloorPlus(Integer.valueOf(ss[pos++]));
      rawLog.setAuctionPrice(Integer.valueOf(ss[pos++]));
      rawLog.setPrice(Integer.valueOf(ss[pos++]));
      rawLog.setPricePlus(Integer.valueOf(ss[pos++]));
      rawLog.setNoTrafficReasons(ss[pos++]);
      rawLog.setBidFailureReasons(ss[pos++]);
      rawLog.setBidRequest(ss[pos++]);
      rawLog.setFileName(filename);
      rawLog.setLineNo(lineno);
    } else {
      throw new ParserException.LogFormatException("log-fileds-not-enough-only-" + ss.length);
    }

    return rawLog;
  }

  public static byte[] converToBinary(RawLog rawLog) throws IOException {
    DatumWriter<RawLog> datumWriter = new SpecificDatumWriter<RawLog>(RawLog.class);
    ByteArrayOutputStream out = new ByteArrayOutputStream();
    Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
    datumWriter.write(rawLog, encoder);
    encoder.flush();
    out.close();
    return out.toByteArray();
  }

  private long getTimestamp(String input) throws ParserException.LogFormatException {
    long ret = 0;
    try {
      ret = Long.parseLong(input);
    } catch (NumberFormatException e) {
      throw new ParserException.LogFormatException("timestamp-failed-to-parseLong-" + input);
    }

    return ret;
  }

  private LogType getLogType(String type) throws ParserException.LogTypeException {
    int iType = 0;
    try {
      iType = Integer.parseInt(type);
    } catch (NumberFormatException e) {
      throw new ParserException.LogTypeException("log-type-failed-to-parseInt-" + type);
    }

    LogType logType = null;
    switch(iType) {
      case 1:
        logType = LogType.BID;
        break;
      case 2:
        logType = LogType.SHOW;
        break;
      case 3:
        logType = LogType.CLICK;
        break;
      default:
        throw new ParserException.LogTypeException("log-type-not-found-" + type);
    }

    return logType;
  }

  private TrafficSource getTrafficSource(String type) throws ParserException.LogFormatException{
    int iType = 0;
    try {
      iType = Integer.parseInt(type);
    } catch (NumberFormatException e) {
      throw new ParserException.LogFormatException("traffice-source-failed-to-parseInt-" + type);
    }

    TrafficSource traffciSourceType = null;
    switch(iType) {
      case 1:
        traffciSourceType = TrafficSource.SITE;
        break;
      case 2:
        traffciSourceType = TrafficSource.APP;
        break;
      default:
        traffciSourceType = TrafficSource.SITE;
    }

    return traffciSourceType;
  }

  private CreativeType getCreativeType(String type) throws ParserException.LogFormatException{
    int iType = 0;
    try {
      iType = Integer.parseInt(type);
    } catch (NumberFormatException e) {
      throw new ParserException.LogFormatException("creative-type-failed-to-parseInt-" + type);
    }

    CreativeType creativeType= null;
    switch(iType) {
      case 1:
        creativeType = CreativeType.BANNER;
        break;
      case 2:
        creativeType = CreativeType.VIDEO;
        break;
      case 3:
        creativeType = CreativeType.AUDIO;
        break;
      case 4:
        creativeType = CreativeType.NATIVE;
        break;
      default:
        creativeType = CreativeType.BANNER;
    }

    return creativeType;
  }

  private Distribution getDistribution(String type) throws ParserException.LogTypeException {
    int iType = 0;
    try {
      iType = Integer.parseInt(type);
    } catch (NumberFormatException e) {
      throw new ParserException.LogTypeException("ditribution-failed-to-parseInt-" + type);
    }

    Distribution distribution = null;
    switch(iType) {
      case 0:
        distribution = Distribution.RTB;
        break;
      case 1:
        distribution = Distribution.ADNETWORK;
        break;
      case 2:
        distribution = Distribution.EXCHANGE;
        break;
      default:
        distribution = Distribution.EXCHANGE;
    }

    return distribution;
  }


  private DistributionType getDistributionType(String type) throws ParserException.LogTypeException {
    int iType = 0;
    try {
      iType = Integer.parseInt(type);
    } catch (NumberFormatException e) {
      throw new ParserException.LogTypeException("distribution-type-failed-to-parseInt-" + type);
    }

    DistributionType distributionType = null;
    switch(iType) {
      case 0:
        distributionType = DistributionType.RTB;
        break;
      case 1:
        distributionType = DistributionType.DEAL;
        break;
      case 2:
        distributionType = DistributionType.TRANSFER;
        break;
      default:
        distributionType = DistributionType.RTB;
    }

    return distributionType;
  }
}
